package com.winit.schedule.job.tracker.remoting;

import io.netty.channel.ChannelHandlerContext;

import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.winit.schedule.core.event.Dispatcher;
import com.winit.schedule.core.model.AssignedTask;
import com.winit.schedule.core.model.TaskID;
import com.winit.schedule.core.protocol.Protocol;
import com.winit.schedule.core.protocol.command.HeartbeatRequest;
import com.winit.schedule.core.protocol.command.HeartbeatResponse;
import com.winit.schedule.core.repository.constants.TaskItemStatus;
import com.winit.schedule.core.repository.entity.TaskTrackerEntity;
import com.winit.schedule.core.task.ScheduleTaskDefine;
import com.winit.schedule.core.utils.Clock;
import com.winit.schedule.job.tracker.event.TaskEventType;
import com.winit.schedule.job.tracker.event.TaskStatusChangeEvent;
import com.winit.schedule.job.tracker.event.persist.DataPersistEvent;
import com.winit.schedule.job.tracker.event.persist.DataPersistEventType;
import com.winit.schedule.job.tracker.repository.ScheduleDataManager;
import com.winit.schedule.job.tracker.schedule.TaskScheduleManager;
import com.winit.schedule.remoting.exception.RemotingCommandException;
import com.winit.schedule.remoting.protocol.RemotingCommand;

/**
 * 心跳处理类
 * 
 * @author jianke.zhang 2015-3-27 下午2:59:46
 * @since 1.0
 */
public class HeartBeatProcessor extends AbstractProcessor {

    private final static Logger       LOGGER = LoggerFactory.getLogger(HeartBeatProcessor.class);

    private final ExecutorService     executor;

    private ScheduleDataManager       scheduleDataManager;

    private Dispatcher                dataDispatcher;

    private final TaskScheduleManager taskScheduleManager;

    public HeartBeatProcessor(RemotingService remotingService){
        super(remotingService);
        this.dataDispatcher = remotingService.getJobTrackerContext().getDataDispatcher();
        this.taskScheduleManager = remotingService.getJobTrackerContext().getTaskScheduleManager();
        this.scheduleDataManager = remotingService.getJobTrackerContext().getScheduleDataManager();
        executor = Executors.newCachedThreadPool();
    }

    @SuppressWarnings("unchecked")
    @Override
    public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand command)
                                                                                             throws RemotingCommandException {

        long startTime = System.currentTimeMillis();
        HeartbeatRequest body = command.getBody();
        if (body == null) {
            return RemotingCommand.createResponseCommand(Protocol.ResponseCode.UNKNOW.code(), "unknow request");
        }

        HeartbeatResponse res = new HeartbeatResponse();

        long endHeartbeatTime = Clock.getSystemTime();
        TaskTrackerEntity taskTracker = body.getTaskTrackerEntity();

        long rpcTimeout = taskTracker.getRpcTimeout();
        long startHeartbeatTime = taskTracker.getHeartbeatTime();
        if (LOGGER.isDebugEnabled()) {
            long usedTime = endHeartbeatTime - startHeartbeatTime;
            LOGGER.debug("From [" + taskTracker.getTaskTrackerUuid() + "]'s heartbeat using time(ms):" + usedTime);
        }

        taskTracker.setRegisterTime(endHeartbeatTime);
        taskTracker.setHeartbeatTime(endHeartbeatTime);
        // 更新缓存信息
        remotingService.getJobTrackerContext().registerTaskTracker(taskTracker);

        this.dataDispatcher.getEventHandler().handle(new DataPersistEvent(DataPersistEventType.REG_TASK_TRACKER,
            taskTracker,
            body.getTaskRunners()));

        // 更新数据库信息
        TaskTrackerEntity oldTaskTracker = scheduleDataManager.getTaskTrackerByUuId(taskTracker.getTaskTrackerUuid());

        // 记录taskTracker连接日志
        if (null != oldTaskTracker) {

            // 更新任务执行状态
            if (null != body.getTaskRunners()) {
                if (LOGGER.isDebugEnabled()) {
                    LOGGER.debug("From " + taskTracker.getTaskTrackerUuid() + " taskRunner size is:"
                                 + body.getTaskRunners().size());
                }

                // 更新线程组信息
                dataDispatcher.getEventHandler()
                    .handle(new DataPersistEvent(DataPersistEventType.TASKITEM_STATUS_UPDATE,
                        body.getTaskRunners(),
                        TaskItemStatus.ACTIVIE));
            }

            checkScheduleTaskDefine(oldTaskTracker.getTaskTrackerUuid(),
                body.getTaskTrackerEntity().getTaskDefines(),
                    oldTaskTracker.getTaskDefines());


            boolean hasBlackList = scheduleDataManager.containBlackList(oldTaskTracker.getIp());

            if (hasBlackList) {
                List<TaskID> taskIds = scheduleDataManager.queryAllTaskIdByTaskTracker(oldTaskTracker.getTaskGroup());
                res.setStopTaskIds(taskIds);
                res.setTasks(Collections.<AssignedTask>emptyList());
            } else {
                // JobTracker启动一个心跳超时时间后才开始分配任务，避免多Tasktracker同一时间连接频繁重新分配任务
                List<AssignedTask> executableTasks = null;
                if (remotingService.getJobTrackerContext().allowAssignedTasks()) {

                    if (LOGGER.isDebugEnabled()) {
                        LOGGER.debug("Began to query and obtain the executable task");
                    }

                    try {
                        executableTasks = taskScheduleManager.getExecutableTask(oldTaskTracker, body.getTaskRunners());
                    } catch (Exception e) {
                        LOGGER.error("Fetch task failure.", e);
                    }
                }
                List<TaskID> stopTaskIds = scheduleDataManager.queryAllStopTaskIDs();
                res.setStopTaskIds(stopTaskIds);
                // 是否获取到可执行任务
                if (null != executableTasks && executableTasks.size() > 0) {
                    if (LOGGER.isDebugEnabled()) {
                        LOGGER.debug("Fetch the executable task size : " + executableTasks.size());
                    }

                    // 异步事件
                    this.remotingService.getJobTrackerContext()
                            .getDispatcher()
                            .getEventHandler()
                            .handle(new TaskStatusChangeEvent(TaskEventType.TASK_PUSH,
                                    executableTasks,
                                    oldTaskTracker.getTaskTrackerUuid()));

                    // 推送可执行任务到taskTracker
                    res.setTasks(executableTasks);
                }
            }
        }

        if (LOGGER.isDebugEnabled()) {
            long processUsedTime = System.currentTimeMillis() - startTime;
            LOGGER.debug("Start response request, Heartbeat event processing time(ms):" + processUsedTime);
        }

        // 重置响应开始时间，用于记录网络请求用时
        res.setTimestamp(Clock.getSystemTime());

        return RemotingCommand.createResponseCommand(Protocol.ResponseCode.SUCCESS.code(),
            "heart beat received success!",
            res);
    }

    /**
     * 检查未注册成功的任务，重新注册。
     * @param taskTrackerUUID
     * @param taskTrackerDefines
     * @param registeredDefines
     */
    private void checkScheduleTaskDefine(String taskTrackerUUID, Set<ScheduleTaskDefine> taskTrackerDefines,
                                         Set<ScheduleTaskDefine> registeredDefines) {
        if (null != taskTrackerDefines && !taskTrackerDefines.isEmpty()) {
            Iterator<ScheduleTaskDefine> iter = taskTrackerDefines.iterator();

            ScheduleTaskDefine taskDefine = null;
            try {
                while (iter.hasNext()) {
                    taskDefine = iter.next();
                    if (null == registeredDefines || !registeredDefines.contains(taskDefine)) {
                        scheduleDataManager.saveScheduleTaskDefine(taskDefine, taskTrackerUUID);
                    }
                }
            } catch (Exception e) {
                LOGGER.warn("[checkScheduleTaskDefine]--> Save scheduleTaskDefine failure.", e.getMessage());
                // ignore e
            }
        }
    }

    public ExecutorService getExecutor() {
        return executor;
    }

}
